package com.flume.ModifyFilter;

import java.io.UnsupportedEncodingException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;

public class AddBodyTimeFilter implements Interceptor {
	
	public static Logger logger = LoggerFactory.getLogger(AddBodyTimeFilter.class);
	
	private String field;
	private String format;
	private String encoding;
	private String formatRaw;
	private String newFiled;
	
	public AddBodyTimeFilter(String field, String formatRaw, String format, String encoding, String newField) {
		this.field = field;
		this.formatRaw = formatRaw;
		this.format = format;
		this.encoding = encoding;
		this.newFiled = newField;
	}
	
	public void initialize() {
		// TODO Auto-generated method stub
		logger.info("origin field :{}, add field :{}.",field,newFiled);
	}

	
	public String parseToDatestr(String format,String formatRaw, String time) {
		
		SimpleDateFormat formater = new SimpleDateFormat(format);
		String dateStr = null;
		try {
			if("timestamp".equals(formatRaw)) {
				dateStr = formater.format(new Date(Long.parseLong(time)));
			}else {
				dateStr = formater.format(formater.parse(time));
			}
			
		} catch(NumberFormatException e) {
			logger.error("parse long exception:{} ",time);
			e.printStackTrace();
		} catch (ParseException e) {
			// TODO Auto-generated catch block
			logger.error("Unparseable date:{} ",time);
			e.printStackTrace();
		}
		return dateStr;
	}
	
	public long parseToTimestamp(String formatRaw,String time) {
		
		long timestamp = 0;
		try {
			SimpleDateFormat formater = new SimpleDateFormat(formatRaw);
			Date date = formater.parse(time);
			timestamp = date.getTime();
		} catch (ParseException e) {
			// TODO Auto-generated catch block
			logger.error("Unparseable date:{} ",time);
			e.printStackTrace();
		}
		return timestamp;
	}
	
	public Event intercept(Event event) {
		// TODO Auto-generated method stub
		if(field==null||"".equals(field)||newFiled==null||"".equals(newFiled)) {
			logger.warn("field or newField not set");
			return event;
		}
		try {
			JSONObject eventBody = JSONObject.parseObject(new String(event.getBody(), encoding));
			String timeValue = eventBody.getString(field);
			if("timestamp".equals(format)) {
				long time = parseToTimestamp(formatRaw, timeValue);
				eventBody.put(newFiled, time);
			}else {
				String time = parseToDatestr(format, formatRaw, timeValue);	
				eventBody.put(newFiled, time);
			}	
			event.setBody(eventBody.toString().getBytes(encoding));
			
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			logger.error("unsupported encoding : {}",encoding);
			e.printStackTrace();			
		} catch (Exception e) {
			logger.error("error: {}",e);
			e.printStackTrace();
		}
		
		
		return event;
	}

	public List<Event> intercept(List<Event> events) {
		// TODO Auto-generated method stub
		List<Event> result = new ArrayList<Event>();
		for(Event e:events) {
			result.add(intercept(e));
		}
		return result;
	}

	public void close() {
		// TODO Auto-generated method stub
		
	}

	/**
	 * 
	 * @author lenovo
	 * @describe field  原始消息中需要截取的时间字段，如为空，则返回原始消息
	 * 			formatRaw 原始消息中时间字段格式，默认为yyyy-mm-dd，目前支持两种格式：日期字符串（2020-10-21 10:30:30）、timestamp(1603098465488)
	 * 			format 转换后的日期格式（常规日期格式或者timestamp），默认为yyyy-MM-dd，如为unixtime，则转换为timestamp格式
	 * 			encoding 字符编码，默认为utf8
	 * 			newField 新增字段名称，如为空，则返回原始消息
	 */
	public static class Builder implements Interceptor.Builder{

		public static String FIELD="field";
		// yyyy-mm-dd / timestamp
		public static String FORMATRAW="formatRaw";  
		public static String FORMAT="format";
		public static String ENCODING="encoding";
		public static String NEWFIELD = "newField";
		
		private String field;
		private String formatRaw;
		private String format;
		private String encoding;
		private String newField;
		
		public void configure(Context context) {
			// TODO Auto-generated method stub
			field = context.getString(FIELD, null);
			formatRaw = context.getString(FORMATRAW,"yyyy-MM-dd");
			format = context.getString(FORMAT, "yyyy-MM-dd");
			encoding = context.getString(ENCODING, "UTF-8");
			newField = context.getString(NEWFIELD,null);
		}

		public Interceptor build() {
			// TODO Auto-generated method stub
			return new AddBodyTimeFilter(field, formatRaw, format, encoding, newField);
		}
		
	}
}
